package com.lvmama.rhino.service.impl;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.lvmama.rhino.pojo.PageMonitorInfo;
import com.lvmama.rhino.utils.Constants;  
  
@Service  
public class ElasticsearchIndexService {  
  
    private static final Logger logger = Logger.getLogger(ElasticsearchIndexService.class);  
  
    /*@Autowired  
    private Client esClient; */
    
    public void init(){  
        
    }  
    
    public TransportClient getClient() {
        Settings settings = Settings.settingsBuilder().put("cluster.name", Constants.getConfig("elasticsearch.cluster.name")).build();
        TransportClient client = null;
        try {
            String[] ips = Constants.getConfig("elasticsearch.host").split(",");
            int port = Integer.parseInt(Constants.getConfig("elasticsearch.port"));
            client = TransportClient.builder().settings(settings).build();
            for(String ip : ips){   
                client = client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), port));
            }
            //client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(Constants.getConfig("elasticsearch.host")), 9300));
        }
        catch (UnknownHostException e) {
            e.printStackTrace();
        }
       return client;
    }
    
    /**
     * 创建索引
     * @throws IOException
     */
    public void create(String indexName, String typeName) throws IOException{
        TransportClient client = getClient();
        IndicesExistsRequest inExistsRequest = new IndicesExistsRequest(indexName);
        IndicesExistsResponse inExistsResponse = client.admin().indices().exists(inExistsRequest).actionGet();
        if (!inExistsResponse.isExists()) {
            client.admin().indices().prepareCreate(indexName).execute().actionGet();
        }
        XContentBuilder mapping = XContentFactory.jsonBuilder()  
            .startObject()  
            .startObject(typeName) 
            .startObject("properties")
            .startObject("province").field("type", "string").field("index", "not_analyzed").endObject() 
            .startObject("city").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("operators").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("category_id").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("network_type").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("load_time").field("type", "double").field("index", "not_analyzed").endObject()
            .startObject("total_load_time").field("type", "double").field("index", "not_analyzed").endObject()
            .startObject("is_lv").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("create_date").field("type", "date").field("index", "not_analyzed").endObject()
            .startObject("visit_times").field("type", "long").field("index", "not_analyzed").endObject()
            .startObject("proportion").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("average_time").field("type", "double").field("index", "not_analyzed").endObject()
            .startObject("average_time_str").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("page_code").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("url").field("type", "string").field("index", "not_analyzed").endObject()
            .startObject("date").field("type", "date").field("index", "not_analyzed").endObject()
           /*.startObject("name").field("type", "string").field("store", "yes").endObject()  
            .startObject("birthday").field("type", "date").field("format", "YYYY-MM-dd").endObject()  
            .startObject("location").field("lat", "double").field("lon", "double").endObject()*/  
            .endObject()  
            .endObject()  
            .endObject();  
        PutMappingRequest mappingRequest = Requests.putMappingRequest(indexName).type(typeName).source(mapping);
        client.admin().indices().putMapping(mappingRequest).actionGet();
        client.close();
    }
    
    /**
     * 删除索引
     */
    public void deleteIndex(String indexName){
        IndicesExistsRequest inExistsRequest = new IndicesExistsRequest(indexName);
        TransportClient client = getClient();
        IndicesExistsResponse inExistsResponse = client.admin().indices().exists(inExistsRequest).actionGet();
        if (inExistsResponse.isExists()) {
            DeleteIndexResponse response = client.admin().indices().prepareDelete(indexName).execute().actionGet();
        } 
        client.close();
    }
  
    
    /**
     * 批量插入数据
     * @param pageMonitorInfoList
     * @return
     */
    public boolean insertOrUpdatePageMonitorInfo(String indexName, String typeName, List<PageMonitorInfo> pageMonitorInfoList) {  
        /*List<IndexQuery> queries = new ArrayList<IndexQuery>();  
        for (PageMonitorInfo pageMonitorInfo : pageMonitorInfoList) {  
            IndexQuery indexQuery = new IndexQueryBuilder().withObject(pageMonitorInfo).build();  
            queries.add(indexQuery);  
        }  
        elasticsearchTemplate.bulkIndex(queries);
        return true; */ 
        TransportClient client = getClient();
        int count = 1;
        BulkRequestBuilder bulkRequest = client.prepareBulk();
        for (PageMonitorInfo pageMonitorInfo : pageMonitorInfoList) { 
            Map<String,Object> map = new HashMap<String,Object>();
            map.put("province", pageMonitorInfo.getProvince());
            map.put("city", pageMonitorInfo.getCity());
            map.put("operators", pageMonitorInfo.getOperators());
            map.put("network_type", pageMonitorInfo.getNetwork_type());
            map.put("total_load_time", pageMonitorInfo.getTotal_load_time());
            map.put("visit_times", pageMonitorInfo.getVisit_times());
            map.put("average_time", pageMonitorInfo.getAverage_time());
            map.put("url", pageMonitorInfo.getUrl());
            map.put("date", pageMonitorInfo.getDate());
            bulkRequest.add(client.prepareIndex(indexName, typeName).setSource(JSON.toJSONString(map)));
            if(count % 10000 == 0){
                bulkRequest.execute().actionGet();
                bulkRequest = client.prepareBulk();
            }
            count ++;
        } 
        bulkRequest.execute().actionGet();
        client.close();
        return true;  
    }  
  
  
    public boolean insertOrUpdatePageMonitorInfo(PageMonitorInfo pageMonitorInfo) {  
        try {  
            IndexQuery indexQuery = new IndexQueryBuilder().withId(null).withObject(pageMonitorInfo).build();  
//            elasticsearchTemplate.index(indexQuery);  
            return true;  
        } catch (Exception e) {  
            logger.error("insert or update task info error.", e);  
            return false;  
        }  
    }  
  
  
    public <T> boolean deleteById(String id, Class<T> clzz) {  
        try {  
//            elasticsearchTemplate.delete(clzz, id);  
            return true;  
        } catch (Exception e) {  
            logger.error("delete " + clzz + " by id " + id + " error.", e);  
            return false;  
        }  
    }  
  
}  
